package com.dingqb.gradle1.rxjava;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class RxjavaBase {
    public static void interval() {

        Observable.interval(1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        System.err.println("onSubscribe");
                    }

                    @Override
                    public void onNext(Long aLong) {
                        System.err.println(aLong);
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.err.println(e.getMessage());


                    }

                    @Override
                    public void onComplete() {
                        System.err.println("complete");
                    }
                });

    }

    public static void intervalcount() {
        Observable.intervalRange(1, 100, 2, 2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.computation())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        System.err.println("onSubscribe");
                    }

                    @Override
                    public void onNext(Long aLong) {
                        System.err.println(aLong);
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.err.println(e.getMessage());


                    }

                    @Override
                    public void onComplete() {
                        System.err.println("complete");
                    }
                });

    }

    public static void just() {
        Observable.create(new ObservableOnSubscribe<Long>() {
            @Override
            public void subscribe(ObservableEmitter<Long> e) throws Exception {
                System.err.println(" subscribe" + Thread.currentThread().getName());
                for (long i = 0; i < 1010; i++) {
                    e.onNext(i);
                }
                e.onComplete();
            }
        })
                .subscribeOn(Schedulers.computation())
                .observeOn(Schedulers.newThread())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        System.err.println(" onSubscribe" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(Long aLong) {
                        System.err.println("onNext===" + aLong + ", " + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.err.println(e.getMessage());


                    }

                    @Override
                    public void onComplete() {
                        System.err.println("complete");
                    }
                });
    }
}
